package com.event.streaming.process.consumer
import com.event.streaming.common.Runnable
import com.event.streaming.common.straming.StreamingConsumer
import com.event.streaming.config.Settings
import com.event.streaming.config.setttings.{DebugSettings, StreamingConsumerSettings}
import com.event.streaming.process.EventPredictor
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.codehaus.jackson.map.deser.std.StringDeserializer

@SerialVersionUID(29259234L)
class TestConsumer extends StreamingConsumer[StringDeserializer,StringDeserializer]("groupTest") with Runnable with Serializable { self: StreamingConsumerSettings =>
  override def topic: String = "test"

  private var eventPredictor:Option[EventPredictor] = None

  private def createPredictor(spark:SparkSession):Option[EventPredictor] = {
    this.eventPredictor match {
      case Some(_) => { }
      case _ => {
        val debug:Option[DebugSettings] = if(this.debugEnabled) Some(this) else None
        this.eventPredictor = Some(new EventPredictor(spark,debug) with Settings.Sink.Settings)
      }
    }
    this.eventPredictor
  }

  private def process(spark:SparkSession,rdd:RDD[(String,String)]):Unit = for (predictor <- this.createPredictor(spark)) {
    predictor.process(rdd.map(r => r._2))
  }


  override def run(spark:SparkSession): Unit = super.start[String, String](spark, process)






}
